package com.sdgs.paas.sdk.core.web.client;

import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.http.HttpMethod;
import org.springframework.http.ResponseEntity;
import org.springframework.http.client.AsyncClientHttpRequestFactory;
import org.springframework.http.client.SimpleClientHttpRequestFactory;
import org.springframework.util.StringUtils;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import com.sdgs.paas.sdk.core.log.PaasLogger;

/**
 * 异步查询帮助类
 *
 * @author wjj
 *
 */

public class PaasAsyncQueryHelperImpl implements IPaasAsyncQueryHelper {

	@Autowired
	private PaasAsyncRestTemplate asyncRestTemplate;

	@Autowired
	@Qualifier("externalAsyncRestTemplate")
	private PaasAsyncRestTemplate externalAsyncRestTemplate;

	@Override
	public <T, K> K asyncQueryForEntity(Map<String, PaasAsyncQueryRequest<T>> paasAsyncQueryRequests, int timeout,
			Function<Map<String, ResponseEntity<T>>, K> callback) throws InterruptedException {

		CountDownLatch latch = new CountDownLatch(paasAsyncQueryRequests.size());
		Map<String, ResponseEntity<T>> resultAll = new HashMap<String, ResponseEntity<T>>();

		for (Entry<String, PaasAsyncQueryRequest<T>> entry : paasAsyncQueryRequests.entrySet()) {
			String taskName = entry.getKey();
			PaasAsyncQueryRequest<T> req = entry.getValue();
			PaasAsyncRestTemplate rest = asyncRestTemplate;

			if (req.getType().equals(PaasAsyncQueryRequest.EXTERNALCALL)) {
				rest = externalAsyncRestTemplate;

			}
			HttpMethod method = HttpMethod.GET;
			if (req.getRequest() != null) {
				method = HttpMethod.POST;
			}

			ListenableFuture<ResponseEntity<T>> resp = null;
			ParameterizedTypeReference<T> repPtype = req.getRepPType();

			if (repPtype != null) {
				resp = rest.exchange(req.getUrl(), method, req.getRequest(), repPtype, req.getUriVariables());
			} else {
				resp = rest.exchange(req.getUrl(), method, req.getRequest(), req.getResponseType(),
						req.getUriVariables());
			}

			String params = StringUtils.arrayToCommaDelimitedString(req.getUriVariables().values().toArray());

			resp.addCallback(new ListenableFutureCallback<ResponseEntity<T>>() {
				@Override
				public void onSuccess(ResponseEntity<T> resp) {
					resultAll.put(taskName, resp);
					PaasLogger.info("服务 {} 调用成功，参数 {} ", req.getUrl(), params);
					latch.countDown();
				}

				@Override
				public void onFailure(Throwable ex) {
					PaasLogger.error(ex, "服务 {} 调用失败，参数 {} ", req.getUrl(), params);
					latch.countDown();
				}
			});

		}
		latch.await(timeout, TimeUnit.SECONDS);

		return callback.apply(resultAll);
	}

	@Override
	public void cfgConcurrencyLimit(int concurrencyLimit) {

		SimpleClientHttpRequestFactory requestFactory = new SimpleClientHttpRequestFactory();
		SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor();
		taskExecutor.setConcurrencyLimit(concurrencyLimit);
		requestFactory.setTaskExecutor(taskExecutor);
		cfgAsyncClientHttpRequestFactory(requestFactory);

	}

	@Override
	public void cfgAsyncClientHttpRequestFactory(AsyncClientHttpRequestFactory asyncRequestFactory) {
		synchronized (asyncRestTemplate) {
			asyncRestTemplate.setAsyncRequestFactory(asyncRequestFactory);
		}
		synchronized (externalAsyncRestTemplate) {
			externalAsyncRestTemplate.setAsyncRequestFactory(asyncRequestFactory);
		}
	}

}
